package com.shujia.opt

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

import java.sql.{Connection, DriverManager, PreparedStatement}

object Demo1ForeachPartition {
  def main(args: Array[String]): Unit = {
    //1、创建spark环境
    val spark: SparkSession = SparkSession
      .builder()
      .appName("fp")
      .master("local[4]")
      .getOrCreate()

    //获取sparkContext环境
    val sc: SparkContext = spark.sparkContext


    //2、读取学生表的数据
    val studentsRDD: RDD[String] = sc.textFile("spark/data/student")



    //3、将数据写入mysql
    studentsRDD.foreachPartition((iter: Iterator[String]) => {
      /**
       * 使用foreachPartition每一个分区创建一个数据库连接
       *
       */

      val start: Long = System.currentTimeMillis()

      //1、加载驱动
      Class.forName("com.mysql.jdbc.Driver")
      //2、创建数据库连接，创建数据库连接大概需要10毫秒
      //不能在算子外创建数据库连接，再算子内去使用，（网络链接不能在网络中传输）算子外是Driver端，算子内是Executor端
      val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")

      val end: Long = System.currentTimeMillis()

      println(s"创建连接用了：${end - start}:毫秒")


      //迭代器的foreach和rdd的foreach是不同的，
      //迭代器的foreach内外都在Executor端
      iter.foreach(stu => {
        //3、编写sql
        val stat: PreparedStatement = con.prepareStatement("insert into student(id,name,age,gender,clazz) values(?,?,?,?,?)")

        val split: Array[String] = stu.split(",")

        //设置数据
        stat.setInt(1, split(0).toInt)
        stat.setString(2, split(1))
        stat.setInt(3, split(2).toInt)
        stat.setString(4, split(3))
        stat.setString(5, split(4))

        //执行sql,插入数据
        stat.execute()
      })

      //关闭连接
      con.close()
    })

  }

}
